#include "soft_types.h"
#include <assert.h>
#include <pthread.h>

#include "ThreadPool/ThreadPool.h"

/**********************************************************
 * Function Name:        ThreadPool_init
 * Description:          初始化线程池
 * Parameters[in]:       INT32 thread_num
                         INT32 queue_max_num
 * Parameters[out]:      void
 * ReturnValue:          THREAD_POOL_s *
 * Author:               liuke
 * Date:                 2020.07.21
***********************************************************/
THREAD_POOL_s *ThreadPool_init(INT32 thread_num, INT32 queue_max_num)
{
    INT32 i = 0;
    THREAD_POOL_s *pool = NULL;
    
    do 
    {
        pool = calloc(1, sizeof(THREAD_POOL_s));
        if (NULL == pool)
        {
            LOG_E("failed to malloc threadpool!\n");
            break;
        }
        
        pool->thread_num = thread_num;
        pool->queue_max_num = queue_max_num;
        pool->queue_cur_num = 0;
        pool->head = NULL;
        pool->tail = NULL;
        if (pthread_mutex_init(&(pool->mutex), NULL))
        {
            LOG_E("failed to init mutex!\n");
            break;
        }
        
        if (pthread_cond_init(&(pool->queue_empty), NULL))
        {
            LOG_E("failed to init queue_empty!\n");
            break;
        }
        
        if (pthread_cond_init(&(pool->queue_not_empty), NULL))
        {
            LOG_E("failed to init queue_not_empty!\n");
            break;
        }
        
        if (pthread_cond_init(&(pool->queue_not_full), NULL))
        {
            LOG_E("failed to init queue_not_full!\n");
            break;
        }
        
        pool->pthreads = calloc(1, sizeof(pthread_t) * thread_num);
        if (NULL == pool->pthreads)
        {
            LOG_E("failed to malloc pthreads!\n");
            break;
        }
        
        pool->queue_close = 0;
        pool->pool_close = 0;
        for (i = 0; i < pool->thread_num; ++i)
        {
            pthread_create(&(pool->pthreads[i]), NULL, ThreadPool_function, (void *)pool);
        }
        
        return pool;    
    } while (0);
    
    return NULL;
}

/**********************************************************
 * Function Name:       ThreadPool_add_job
 * Description:         增加任务
 * Parameters[in]:      DB_POOL_s *pool
                        POOL_CB_F callback
                        void *arg
 * Parameters[out]:     void
 * ReturnValue:         INT32
 * Author:              liuke
 * Date:                2020.07.21
***********************************************************/
INT32 ThreadPool_add_job(THREAD_POOL_s *pool, THREAD_POOL_CB callback, void *arg)
{
    assert(pool != NULL);
    assert(callback != NULL);
    assert(arg != NULL);

    pthread_mutex_lock(&(pool->mutex));
    while ((pool->queue_cur_num == pool->queue_max_num) && !(pool->queue_close || pool->pool_close))
    {
        pthread_cond_wait(&(pool->queue_not_full), &(pool->mutex));
    }

    if (pool->queue_close || pool->pool_close)
    {
        pthread_mutex_unlock(&(pool->mutex));
        return -1;
    }
    
    THREAD_POOL_JOB_s *pjob =(THREAD_POOL_JOB_s *) malloc(sizeof(THREAD_POOL_JOB_s));
    if (NULL == pjob)
    {
        pthread_mutex_unlock(&(pool->mutex));
        return -1;
    }
    
    pjob->callback = callback;    
    pjob->arg = arg;
    pjob->next = NULL;
    if (pool->head == NULL)   
    {
        pool->head = pool->tail = pjob;
        pthread_cond_broadcast(&(pool->queue_not_empty));
    }
    else
    {
        pool->tail->next = pjob;
        pool->tail = pjob;    
    }
    
    pool->queue_cur_num++;
    pthread_mutex_unlock(&(pool->mutex));
    
    return 0;
}

/**********************************************************
 * Function Name:        ThreadPool_function
 * Description:          
 * Parameters[in]:       void *arg
 * Parameters[out]:      void
 * ReturnValue:          void* 
 * Author:               liuke
 * Date:                 2020.07.21
***********************************************************/
void* ThreadPool_function(void* arg)
{
    THREAD_POOL_s *pool = (THREAD_POOL_s *)arg;
    THREAD_POOL_JOB_s *pjob = NULL;
    
    while (1)
    {
        pthread_mutex_lock(&(pool->mutex));
        while ((pool->queue_cur_num == 0) && !pool->pool_close)
        {
            pthread_cond_wait(&(pool->queue_not_empty), &(pool->mutex));
        }

        if (pool->pool_close)
        {
            pthread_mutex_unlock(&(pool->mutex));
            pthread_exit(NULL);
        }
        
        pool->queue_cur_num--;
        pjob = pool->head;
        if (pool->queue_cur_num == 0)
        {
            pool->head = pool->tail = NULL;
        }
        else 
        {
            pool->head = pjob->next;
        }
        
        if (pool->queue_cur_num == 0)
        {
            pthread_cond_signal(&(pool->queue_empty));
        }
        
        if (pool->queue_cur_num == pool->queue_max_num - 1)
        {
            pthread_cond_broadcast(&(pool->queue_not_full));
        }
        
        pthread_mutex_unlock(&(pool->mutex));

        (*(pjob->callback))(pjob->arg);
        free(pjob);
        pjob = NULL;    
    }
}

/**********************************************************
 * Function Name:        ThreadPool_destroy
 * Description:          
 * Parameters[in]:       THREAD_POOL_s *pool
 * Parameters[out]:      void
 * ReturnValue:          INT32,
 * Author:               liuke
 * Date:                 2020.07.21
***********************************************************/
INT32 ThreadPool_destroy(THREAD_POOL_s *pool)
{
    INT32 i;
    THREAD_POOL_JOB_s *p;
    
    assert(pool != NULL);
    pthread_mutex_lock(&(pool->mutex));

    if (pool->queue_close || pool->pool_close)  
    {
        pthread_mutex_unlock(&(pool->mutex));
        return -1;
    }

    pool->queue_close = 1;
    while (pool->queue_cur_num != 0)
    {
        pthread_cond_wait(&(pool->queue_empty), &(pool->mutex));
    }    

    pool->pool_close = 1;      
    pthread_mutex_unlock(&(pool->mutex));
    pthread_cond_broadcast(&(pool->queue_not_empty));
    pthread_cond_broadcast(&(pool->queue_not_full));

    for (i = 0; i < pool->thread_num; ++i)
    {
        pthread_join(pool->pthreads[i], NULL);
    }

    pthread_mutex_destroy(&(pool->mutex));          
    pthread_cond_destroy(&(pool->queue_empty));
    pthread_cond_destroy(&(pool->queue_not_empty));   
    pthread_cond_destroy(&(pool->queue_not_full));    
    free(pool->pthreads);
    
    while (pool->head != NULL)
    {
        p = pool->head;
        pool->head = p->next;
        free(p);
    }
	
    free(pool);
    return 0;
}

